package com.api.config;

import com.alibaba.fastjson2.JSONObject;
import com.api.entity.Mailbox;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;

/**
 *
 * SSE服务类
 *
 * @Author: yangsf
 * @CreateDate: 2024/10/21
 */

@Slf4j
@Component
public class SseEmitterServer {

    private static final ConcurrentHashMap<String, SseEmitter> SSE_POOL = new ConcurrentHashMap<String, SseEmitter>();
    private static final ConcurrentHashMap<String, Timer> SSE_HEADER_POOL = new ConcurrentHashMap<String, Timer>();

    /**
     * 建立连接
     */
    public SseEmitter connect(String userId){
        log.info("******************开始建立连接*****************");
        String  userCode = RedisConstant.SSE_USER_CODE + userId;
        //设置超时时间，0表示不过期，默认是30秒，超过时间未完成会抛出异常
        SseEmitter sseemitter = new SseEmitter(0L);
        //注册回调
        sseemitter.onCompletion(completionCallBack(userId));
        sseemitter.onError(errorCallBack(userId));
        sseemitter.onTimeout(timeoutCallBack(userId));
        SSE_POOL.put(userCode, sseemitter);
        // 开启心跳活跃
        startHeartbeat(sseemitter,userId);
        return sseemitter;
    }

    /**
     * 推送消息
     */
    public void sendMsg(Long userId, Mailbox mailbox){
        String userCode = RedisConstant.SSE_USER_CODE + userId;
        SseEmitter sse = SSE_POOL.get(userCode);
        if (sse != null) {
            try {
                // 发送事件
                sse.send(JSONObject.toJSONString(mailbox));
            }catch (Exception e){
                log.error("连接信息: {}, 错误消息: {}", userId, e.getMessage());
            }
        } else {
            log.warn("用户 {} 的 SSE 连接不存在。", userId);
        }
    }

    /**
     * 开启心跳
     */
    public void startHeartbeat(SseEmitter sse, String userId) {
        Timer heartbeatTimer = new Timer();
        SSE_HEADER_POOL.put(userId,heartbeatTimer);
        heartbeatTimer.schedule(new TimerTask() {
            @Override
            public void run() {
                if (Objects.nonNull(SSE_HEADER_POOL.get(userId))) {
                    // 发送心跳:保持长连接
                    System.out.println("发送心跳:保持长连接  ==> userId: "+userId);
                    try {
                        long count = Mailbox.create().where(Mailbox::getUserId).eq(userId).and(Mailbox::getStatus).eq(1).count();
                        Mailbox mailbox = Mailbox.create().setKind(0).setCreateTime(new Date()).setMailboxCount(count);
                        sse.send(JSONObject.toJSONString(mailbox));
                    } catch (Exception e) {
                        log.error("connect active error");
                    }
                }
            }
        }, 25000, 25000);
    }

    /**
     * 关闭当前连接
     */
    public void complete(String userCode, String userId){
        SseEmitter sse = SSE_POOL.get(userCode);
        if (sse != null){
            sse.complete();
        }
    }

    /**
     * 关闭所有连接

    public void completeAll(){
        if(!SSEConstant.SSE_EMITTER_POOL.isEmpty()){
            for (SseEmitter sse : SSEConstant.SSE_EMITTER_POOL.values()) {
                if(null != sse){
                    sse.complete();
                }
            }
            SSEConstant.SSE_EMITTER_POOL.clear();
        }
    }
     */


    private  Runnable completionCallBack(String userId) {
        return () -> {
            removeUser(userId);
            log.info("结束连接: {}", userId);
        };
    }

    private  Runnable timeoutCallBack(String userId){
        return ()->{
            removeUser(userId);
            log.error("连接超时: {}", userId);
        };
    }

    private Consumer<Throwable> errorCallBack(String userId){
        return throwable -> {
            removeUser(userId);
            log.error("连接异常: {}",userId);
        };
    }



    private void removeUser(String userId){
        String userCode = RedisConstant.SSE_USER_CODE + userId;
        try {
            SSE_POOL.remove(userCode);
            // 关闭心跳
            stopHeartbeat(userId);
        }catch (Exception e){
            log.error("关闭连接异常{}",e.getMessage());
        }
    }

    /**
     * 关闭心跳
     * @param userId
     */
    public void stopHeartbeat(String userId) {
        Timer timer = SSE_HEADER_POOL.get(userId);
        if (timer!= null){
            timer.cancel();
        }
        SSE_HEADER_POOL.remove(userId);
    }
}
